package com.linys.scala.qf.day06_spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ExeciseRdd_01 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Test").setMaster("local")
    val sc = new SparkContext(conf)

    //通过并行化生成rdd
    val rdd = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 10))

    //map:对rdd里面每一个元乘以2然后排序
    val rdd2: RDD[Int] = rdd.map(_ * 2).sortBy(x => x, true)
    val rdd22: RDD[Int] = rdd.map(_ * 2).sortBy(x => x, false)
    //collect以数组的形式返回数据集的所有元素(是Action算子)
    println(rdd2.collect().toBuffer)
    println(rdd22.collect().toBuffer)

    //filter:该RDD由经过func函数计算后返回值为true的输入元素组成
    val rdd3: RDD[Int] = rdd2.filter(_ > 8)
    println(rdd3.collect().toBuffer)

    val rdd4 = sc.parallelize(Array("a b c", "b c d"))
    //flatMap:将rdd4中的元素进行切分后压平
    val rdd5: RDD[String] = rdd4.flatMap(_.split(" "))
    println(rdd5.collect().toBuffer)
    //假如: List(List("a b", "b c"),List("e c", " i o"))
    val rdd44 = sc.parallelize(List(List("a b", "b c"),List("e c", " i o")))
    //压平 flatMap(_.flatMap(_.split(" ")))
    println(rdd44.flatMap(_.flatMap(_.split(" "))).collect().toBuffer)

    //sample随机抽样
    //withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样
    //fraction抽样比例例如30% 即0.3 但是这个值是一个浮动的值不准确
    //seed用于指定随机数生成器种子 默认参数不传
    val rdd0 = sc.parallelize(1 to 10)
    val sample = rdd0.sample(false, 0.5)
    println(sample.collect().toBuffer)

    //union:求并集
    val rdd6 = sc.parallelize(List(5, 6, 7, 8))
    val rdd7 = sc.parallelize(List(1, 2, 5, 6))
    val rdd8 = rdd6 union rdd7
    println(rdd8.collect.toBuffer)

    //intersection:求交集
    val rdd9 = rdd6 intersection rdd7
    println(rdd9.collect.toBuffer)

    //distinct:去重出重复
    println(rdd8.distinct.collect.toBuffer)

    //join相同的key会被合并
    val rdd10_1 = sc.parallelize(List(("tom", 111), ("jerry", 3), ("kitty", 2)))
    val rdd10_2 = sc.parallelize(List(("jerry", 2), ("tom", 2), ("dog", 10)))
    val rdd10_3 = rdd10_1 join rdd10_2
    println(rdd10_3.collect().toBuffer)

    //左连接和右连接
    //除基准值外是Option类型,因为可能存在空值所以使用Option
    val rdd10_4 = rdd10_1 leftOuterJoin rdd10_2 //以左边为基准没有是null
    val rdd10_5 = rdd10_1 rightOuterJoin rdd10_2 //以右边为基准没有是null
    println(rdd10_4.collect().toList)
    println(rdd10_5.collect().toBuffer)

    val rdd11_1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)))
    val rdd11_2 = sc.parallelize(List(("jerry", 2), ("tom", 2), ("dog", 10)))
    //笛卡尔积
    val rdd11_3 = rdd11_1 cartesian rdd11_2
    println(rdd11_3.collect.toBuffer)

    val rdd11_4 = rdd11_1 union rdd11_2
    println(rdd11_4.collect().toList)
    //根据传入的参数进行分组
    val rdd11_5_1 = rdd11_4.groupBy(_._1)
    println("-----------------------------------")
    println(rdd11_5_1.collect().toList)

    //按照相同key进行分组,并且可以制定分区
    val rdd11_5_2 = rdd11_4.groupByKey
    println("-----------------------------------")
    println(rdd11_5_2.collect().toList)

    //根据相同key进行分组[分组的话需要二元组]
    //cogroup 和 groupBykey的区别
    //cogroup不需要对数据先进行合并就以进行分组 得到的结果是 同一个key 和不同数据集中的数据集合
    //groupByKey是需要先进行合并然后在根据相同key进行分组
    val rdd11_6: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd11_1 cogroup rdd11_2
    println(rdd11_6)


    println(rdd11_4.reduceByKey(_+_).collect.toBuffer)
    println(rdd11_4.groupByKey().collect.toBuffer)


  }

}
